package com.zilue.hfqueue.queue;

import com.zilue.hfqueue.HFQueueConfig;
import com.zilue.hfqueue.queue.block.HFQueueReadBlock;
import com.zilue.hfqueue.queue.block.HFQueueWriteBlock;
import com.zilue.hfqueue.queue.index.HFQueueIndex;

import java.io.IOException;
import java.util.AbstractQueue;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by ZD on 2017/5/3.
 */
public class HFQueue extends AbstractQueue<byte[]> {
    private String topic;
    private HFQueueIndex index;
    private HFQueueReadBlock readBlock;
    private HFQueueWriteBlock writeBlock;
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private AtomicInteger size;

    public static HFQueue.Builder builder() {
        return new Builder();
    }

    public static class Builder {
        private HFQueue queue = new HFQueue();

        public Builder topic(String topic) throws IOException {
            queue.init(topic);
            return this;
        }

        public HFQueue build() {
            return queue;
        }
    }

    private void init(String topic) throws IOException {
        this.topic = topic;
        index = new HFQueueIndex(HFQueueConfig.getQueueFilePath(), topic);
        int writeBlockNumber = index.getWrite_block().intValue();
        int writeBlockIndex = index.getWrite_index().intValue();
        writeBlock = new HFQueueWriteBlock(HFQueueConfig.getQueueFilePath(), topic, writeBlockNumber, writeBlockIndex);

        int readBlockNumber = index.getRead_block().intValue();
        int readBlockIndex = index.getRead_index().intValue();
        readBlock = new HFQueueReadBlock(HFQueueConfig.getQueueFilePath(), topic, readBlockNumber, readBlockIndex);
        size = new AtomicInteger(index.getWrite_number().intValue() - index.getRead_number().intValue());
    }

    private void writeNextBlock() throws IOException {
        int nextBlockNumber = index.getWrite_block().intValue() + 1;
        writeBlock.close();
        writeBlock = new HFQueueWriteBlock(HFQueueConfig.getQueueFilePath(), topic, nextBlockNumber, 0);
        index.updateWriteBlock(nextBlockNumber);
        index.updateWriteIndex(0);
    }

    private void readNextBlock() throws IOException {
        int nextBlockNumber = index.getRead_block().intValue() + 1;
        readBlock.close();
        readBlock = new HFQueueReadBlock(HFQueueConfig.getQueueFilePath(), topic, nextBlockNumber, 0);
        index.updateReadBlock(nextBlockNumber);
        index.updateReadIndex(0);
    }

    @Override
    public Iterator<byte[]> iterator() {
        throw new UnsupportedOperationException();
    }

    @Override
    public int size() {
        return this.size.intValue();
    }

    @Override
    public boolean offer(byte[] bytes) {
        lock.writeLock().lock();
        try {
            int writePostion = writeBlock.writeData(bytes);
            if (writePostion < 0) {
                writeNextBlock();
                writePostion = writeBlock.writeData(bytes);
            }
            index.updateWriteIndex(writePostion);
            index.updateWriteNumber(index.getWrite_number().intValue() + 1);
            int nowSize = size.intValue();
            while (!size.compareAndSet(nowSize, nowSize + 1)) {
                nowSize = size.intValue();
            }
            return true;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public byte[] peek() {
        lock.readLock().lock();
        try {
            if (index.getRead_block().intValue() == index.getWrite_block().intValue() && index.getRead_index().intValue() == index.getWrite_index().intValue()) {
                throw new IndexOutOfBoundsException();
            }
            int read_index = index.getRead_index().intValue();
            byte[] bytes = readBlock.readData(read_index);
            if (bytes == null) {
                readNextBlock();
            }
            bytes = readBlock.readData(read_index);
            return bytes;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            lock.readLock().unlock();
        }
    }


    @Override
    public byte[] poll() {
        lock.writeLock().lock();
        try {
            if (index.getRead_block().intValue() >= index.getWrite_block().intValue() && index.getRead_index().intValue() >= index.getWrite_index().intValue()) {
                throw new IndexOutOfBoundsException();
            }
            int read_index = index.getRead_index().intValue();
            byte[] bytes = readBlock.readData(read_index);
            if (bytes == null) {
                readNextBlock();
                read_index = 0;
                bytes = readBlock.readData(read_index);
            }
            index.updateReadIndex(read_index + bytes.length + HFQueueReadBlock.INT_SIZE);
            index.updateReadNumber(index.getRead_number().intValue() + 1);
            int nowSize = size.intValue();
            while (!size.compareAndSet(nowSize, nowSize - 1)) {
                nowSize = size.intValue();
            }
            System.out.println(Thread.currentThread().getId() + "|" + read_index + "|" + new String(bytes, "utf-8"));
            return bytes;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            lock.writeLock().unlock();
        }
    }

}
